package com.rrd.dw.mr.gzip.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ucredit.babel.common.model.BaseModel;
import com.ucredit.babel.common.model.RecordType;
import com.ucredit.babel.common.model.third.baihang.BaihangRIDS0000Record;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper.Context;

import java.io.IOException;

/**
 * 百行-信贷行为标签
 *
 * @author guoqiang
 */
public class BaihangRIDS0000Parse extends ParseBase {
    private Text val = new Text();
    private Text key = new Text();

    /**
     * 百行-信贷行为标签
     *
     * @param model
     * @throws InterruptedException
     * @throws IOException
     */
    public boolean parseJob(Context context, BaseModel<?> model, String successful) throws IOException, InterruptedException {
        BaihangRIDS0000Record obj = (BaihangRIDS0000Record) model;
        StringBuilder sb = getBaseInfo(obj, successful);
        String data = obj.getData();
        JSONObject dataJson = JSON.parseObject(data);
        JSONObject head = dataJson.getJSONObject("head");
        sb.append(formatValue(head.getString("requestRefId"))).append(CRT001);
        sb.append(formatValue(head.getString("responseRefId"))).append(CRT001);
        sb.append(formatValue(head.getString("result"))).append(CRT001);
        sb.append(formatValue(head.getString("responseCode"))).append(CRT001);
        sb.append(formatValue(head.getString("responseMsg"))).append(CRT001);

        JSONObject response = dataJson.getJSONObject("response");
        JSONObject cert = response.getJSONObject("cert");
        sb.append(formatValue(cert.getString("idLoanActiveNum"))).append(CRT001);
        sb.append(formatValue(cert.getString("idLoan30dNum"))).append(CRT001);
        sb.append(formatValue(cert.getString("idLoan90dNum"))).append(CRT001);
        sb.append(formatValue(cert.getString("idLoan180dNum"))).append(CRT001);
        sb.append(formatValue(cert.getString("idLoan360dNum"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOrg30dNum"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOrg90dNum"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOrg180dNum"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOrg360dNum"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOrgActiveNum"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOverdueLoanAmt"))).append(CRT001);
        sb.append(formatValue(cert.getString("idActiveOverdueLoanBal"))).append(CRT001);
        sb.append(formatValue(cert.getString("idActiveOverdueLoanNum"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOverdueLoanNum30"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOverdueLoanNum60"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOverdueLoanNum90"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOverdueLoanNum120"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOverdueLoanNum150"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOverdueLoanNum180"))).append(CRT001);
        sb.append(formatValue(cert.getString("idTotalOverdueLoanNum"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOverdueLoanAmt30"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOverdueLoanAmt60"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOverdueLoanAmt90"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOverdueLoanAmt120"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOverdueLoanAmt150"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOverdueLoanAmt180"))).append(CRT001);
        sb.append(formatValue(cert.getString("idActiveOverdueOrgNum"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOverdueOrgNum30"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOverdueOrgNum90"))).append(CRT001);
        sb.append(formatValue(cert.getString("idOverdueOrgNum180"))).append(CRT001);
        sb.append(formatValue(cert.getString("idTotalOverdueOrgNum"))).append(CRT001);
        sb.append(formatValue(cert.getString("idMaxOverdueDays_12M"))).append(CRT001);
        sb.append(formatValue(cert.getString("idMaxOverdueDays_24M"))).append(CRT001);
        sb.append(formatValue(cert.getString("idMaxOverdueDays_36M"))).append(CRT001);
        sb.append(formatValue(cert.getString("idMaxFirstTermOverdueDays_24M"))).append(CRT001);
        sb.append(formatValue(cert.getString("idMaxOvedueAmt30D_36M"))).append(CRT001);

        JSONObject mobile = response.getJSONObject("mobile");
        sb.append(formatValue(mobile.getString("mobileLoanActiveNum"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileLoan30dNum"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileLoan90dNum"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileLoan180dNum"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileLoan360dNum"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOrg30dNum"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOrg90dNum"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOrg180dNum"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOrg360dNum"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOrgActiveNum"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOverdueLoanAmt"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileActiveOverdueLoanBal"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileActiveOverdueLoanNum"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOverdueLoanNum30"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOverdueLoanNum60"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOverdueLoanNum90"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOverdueLoanNum120"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOverdueLoanNum150"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOverdueLoanNum180"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileTotalOverdueLoanNum"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOverdueLoanAmt30"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOverdueLoanAmt60"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOverdueLoanAmt90"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOverdueLoanAmt120"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOverdueLoanAmt150"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOverdueLoanAmt180"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileActiveOverdueOrgNum"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOverdueOrgNum30"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOverdueOrgNum90"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileOverdueOrgNum180"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileTotalOverdueOrgNum"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileMaxOverdueDays_12M"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileMaxOverdueDays_24M"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileMaxOverdueDays_36M"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileMaxFirstTermOverdueDays_24M"))).append(CRT001);
        sb.append(formatValue(mobile.getString("mobileMaxOvedueAmt30D_36M"))).append(CRT001);
        key.set(getKey());
        val.set(sb.toString().replace("\n", " ").replace("\r", " "));
        context.write(key, val);
        return true;
    }

    @Override
    public String getTableName() {
        return "kafka_babel_baihang_rids_inc";
    }

    @Override
    public String getType() {
        return RecordType.BAIHANG_RIDS0000_RECORD.toString();
    }

    @Override
    public String getComment() {
        return "百行-信贷行为标签";
    }
}
